package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.atguigu.gmall.realtime.app.func.DimSink;
import com.atguigu.gmall.realtime.app.func.MyCustomSchema;
import com.atguigu.gmall.realtime.app.func.TableProcessFunction;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import edu.umd.cs.findbugs.annotations.Nullable;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * Author: Felix
 * Date: 2021/11/26
 * Desc: 业务数据动态分流
 */
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 设置流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        /*
        //TODO 2.检查点相关设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置取消job之后是否保留检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        //2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall/ck1"));
        //2.6 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */

        //TODO 3.从Kafka中读取数据
        //3.1 声明消费主题以及消费者组
        String topic = "ods_base_db_m";
        String groupId = "base_db_app_group";

        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);

        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);

        //TODO 4.对流中数据类型进行转换    jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);
//        jsonObjDS.print("js***");

        //TODO 5.对流中数据 进行简单的ETL
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.filter(
                new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObj) throws Exception {
                        boolean flag =
                                jsonObj.getString("table") != null &&
                                        jsonObj.getString("table").length() > 0 &&
                                        jsonObj.getJSONObject("data") != null &&
                                        jsonObj.getString("data").length() > 3;
                        return flag;
                    }
                }
        );

        //filterDS.print(">>>>");

        //TODO 6.使用FlinkCDC读取配置表数据
        //6.1 创建sourceFunction
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall2020_realtime")
                .tableList("gmall2020_realtime.table_process")
                .username("root")
                .password("050994")
                .startupOptions(StartupOptions.initial())
                .deserializer(new MyCustomSchema())
                .build();

        //6.2 读取配置数据 封装为流
        DataStreamSource<String> mySQLDS = env.addSource(sourceFunction);

        //TODO 7.将配置流数据进行广播
        //7.1 声明广播状态描述器     Key:表名 + 操作类型   Value:配置信息TableProcess对象
        MapStateDescriptor<String, TableProcess> mapStateDescriptor
                = new MapStateDescriptor<String, TableProcess>("mapStateDescriptor",String.class,TableProcess.class);
        //7.2 向下广播
        BroadcastStream<String> broadcastDS = mySQLDS.broadcast(mapStateDescriptor);

        //TODO 8.将主流（业务数据流）和广播流(配置数据流) 进行连接
        BroadcastConnectedStream<JSONObject, String> connectDS = filterDS.connect(broadcastDS);



        //TODO 9.动态分流的实现    维度---维度侧输出流     实时---主流
        //9.1 定义维度侧输出流标签
        OutputTag<JSONObject> dimTag = new OutputTag<JSONObject>("dimTag"){};

        //9.2 动态分流
        SingleOutputStreamOperator<JSONObject> realDS = connectDS.process(new TableProcessFunction(dimTag, mapStateDescriptor));

        //9.3 获取维度侧输出流数据
        DataStream<JSONObject> dimDS = realDS.getSideOutput(dimTag);

        realDS.print("事实");
        dimDS.print("维度");


     //TODO 10.将维度侧输出流的数据写到phoenix表汇总
        dimDS.addSink(new DimSink());

        //TODO 11.将主流事实数据写到kafka的主题中
        realDS.addSink(
                MyKafkaUtil.getKafkaSinkBySchema(
                        new KafkaSerializationSchema<JSONObject>() {
                            @Override
                            public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObj, @Nullable Long timestamp) {
                                //获取写到kafka的哪个主题
                                String topic = jsonObj.getString("sink_table");
                                return new ProducerRecord<byte[], byte[]>(topic,jsonObj.getJSONObject("data").toJSONString().getBytes());
                            }
                        }
                )
        );


        env.execute();
    }
}